import asyncio
import json
import logging
import time
from collections import OrderedDict

import oss2
from asgiref.sync import sync_to_async
from channels.generic.websocket import AsyncJsonWebsocketConsumer
from channels.layers import get_channel_layer
from django.utils import timezone
import webrtcvad

from Interview.service.realtime_stt import TingwuRealtimeSTT
from Interview.models import GeneratedQuestion, Answer, Question, Interview, Report, AnswerAnalysis
from Interview.service.report import generate_interview_report
from Interview.service.xf_chat import _generate_ai_question, analyze_answer_text
from InterviewAgent import settings


class InterviewConsumer(AsyncJsonWebsocketConsumer):
    VAD_MODE = 3
    VAD_FRAME_MS = 20
    VAD_SILENCE_SEC = 6.0
    STT_SILENCE_TIMEOUT = 30.0
    SAMPLE_RATE = 16000
    TOTAL_QUESTIONS = 5

    STATIC_FALLBACK_QUESTIONS = [
        "谈谈你印象最深刻的项目经历。",
        "为什么想加入我们公司？",
        "你的职业规划是什么？",
    ]


    async def connect(self):
        self.interview_id = self.scope["url_route"]["kwargs"]["interview_id"]
        # 初始化题量和已发计数
        self.total_questions = self.TOTAL_QUESTIONS
        self.sent_questions = 0
        self.interview_ended = False  # 面试是否已经结束
        self.logger = logging.getLogger(f"InterviewConsumer.{self.interview_id}")
        await self.channel_layer.group_add(f"interview_{self.interview_id}", self.channel_name)
        await self.accept()
        self._reset_state(init=True)
        # 只在 DB 里写首题，等前端主动 REQUEST_QUESTION
        first_q = "请介绍一下你自己。"
        gq = await sync_to_async(GeneratedQuestion.objects.create)(
            interview_id=self.interview_id,
            question_content=first_q,
            question_type="behavioral",
            generation_source=json.dumps({"static": True}, ensure_ascii=False)
        )
        self.current_gq_id = gq.generated_q_id
        self.current_question = first_q
        self.current_qtype = "behavioral"
        self.current_meta = {"static": True}
        self.use_ai_next = True  # False=题库，True=AI

    async def receive(self, text_data=None, bytes_data=None, **_):
        if text_data:
            msg = json.loads(text_data)
            event = msg.get("event")
            # 前端主动请求发当前题（首题或后续题）
            if event == "REQUEST_QUESTION":
                await self.send_json({
                    "event": "QUESTION",
                    "text": f"{self.current_question}",
                    "qtype": self.current_qtype if hasattr(self, 'current_qtype') else "behavioral",
                    "meta": getattr(self, 'current_meta', {}),
                })
                self.logger.info("收到 REQUEST_QUESTION，已发送当前题")
                return
            # --- 1. START_STT → 建立 STT 并发首题 ---
            if event == "START_STT":
                # 新建 STT
                cfg = settings.ALIYUN_TINGWU_CONFIG
                self.stt = TingwuRealtimeSTT(
                    app_key=cfg["APP_KEY"],
                    access_key_id=cfg["ACCESS_KEY_ID"],
                    access_key_secret=cfg["ACCESS_KEY_SECRET"],
                    region=cfg.get("REGION", "cn-beijing"),
                )
                self.stt.main_loop = asyncio.get_running_loop()
                self._bind_stt_callbacks()

                ok = await self.stt.connect()
                if not ok:
                    await self.send_json({"event": "ERROR", "message": "STT连接失败"})
                    return

            # --- 2. QUESTION_READ_COMPLETE → 真正启动录音 & 检测 ---
            if event == "QUESTION_READ_COMPLETE":
                # 重置状态以确保干净
                self._reset_state()
                # 预激活静音帧
                frame = b"\x00" * int(self.SAMPLE_RATE * self.VAD_FRAME_MS / 1000 * 2)
                for _ in range(5):
                    await self.stt.send_audio(frame, throttle=False)
                # 启动两个检测任务
                self.vad_task = asyncio.create_task(self._vad_auto_end_detector())
                self.stt_task = asyncio.create_task(self._stt_auto_end_detector())
                self.logger.info("收到朗读完成，开始 VAD & STT 检测")
                return
            # —— 结束面试 —— #
            if event == "END_INTERVIEW":
                self.interview_ended = True
                self.logger.info("收到 END_INTERVIEW，开始结束流程")

                # 立即终止任何进行中的任务
                if hasattr(self, 'stt') and self.stt.connected:
                    try:
                        await self.stt.close()
                    except Exception as e:
                        self.logger.error(f"关闭STT时出错: {e}")

                # 取消检测任务
                for task_name in ["vad_task", "stt_task"]:
                    if hasattr(self, task_name):
                        task = getattr(self, task_name)
                        if not task.done():
                            task.cancel()

                # 更新面试结束时间
                await sync_to_async(Interview.objects.filter(interview_id=self.interview_id)
                                    .update)(end_time=timezone.now())

                # 通知前端面试结束
                await self.send_json({
                    "event": "INTERVIEW_COMPLETE",
                    "message": "面试已结束，报告生成中..."
                })
                # 启动后台报表流程
                asyncio.create_task(self._run_report())
                return

        # 音频帧推送
        if bytes_data and not self._ended and hasattr(self, 'stt') and self.stt.connected:
            now = time.time()
            for off in range(0, len(bytes_data), self.frame_bytes):
                frame = bytes_data[off: off + self.frame_bytes]
                if len(frame) < self.frame_bytes:
                    frame += b"\x00" * (self.frame_bytes - len(frame))
                try:
                    speech = self.vad.is_speech(frame, sample_rate=self.SAMPLE_RATE)
                except webrtcvad.Error:
                    speech = True
                self.vad_silence_start = None if speech else (self.vad_silence_start or now)
            await self.stt.send_audio(bytes_data)

    def _bind_stt_callbacks(self):
        self.vad = webrtcvad.Vad(self.VAD_MODE)
        self.frame_bytes = int(self.SAMPLE_RATE * self.VAD_FRAME_MS / 1000 * 2)
        self.segments = []
        # 保留主 loop
        self.loop = asyncio.get_running_loop()

        def on_transcript(text, is_final):
            if self._ended:
                return
            self.last_transcript_time = time.time()
            # 明确投递到主 loop
            asyncio.run_coroutine_threadsafe(
                self.send_json({"event": "TRANSCRIPT", "text": text, "is_final": is_final}),
                self.loop
            )
            if is_final:
                ts = timezone.now()
                self.segments.append((ts, text))
                if not self._collect_done and (
                    (self.vad_silence_start and time.time() - self.vad_silence_start > self.VAD_SILENCE_SEC)
                    or (self.segments and time.time() - self.last_transcript_time > self.STT_SILENCE_TIMEOUT)
                ):
                    self._collect_done = True
                    asyncio.run_coroutine_threadsafe(self._end_and_save(), self.loop)

        def on_error(raw, *args):
            if self._ended:
                return
            err_msg = raw.get("raw") if isinstance(raw, dict) else str(raw)
            asyncio.run_coroutine_threadsafe(
                self.send_json({"event": "ERROR", "message": err_msg}),
                self.loop
            )

        self.stt.on_transcript = on_transcript
        self.stt.on_error = on_error

    async def _vad_auto_end_detector(self):
        try:
            while not self._ended:
                await asyncio.sleep(0.2)
                if self.vad_silence_start and time.time() - self.vad_silence_start > self.VAD_SILENCE_SEC:
                    self.logger.info("VAD检测到静音，结束本题回答")
                    return await self._end_and_save()
        except asyncio.CancelledError:
            pass

    async def _stt_auto_end_detector(self):
        try:
            while not self._ended:
                await asyncio.sleep(1)
                if self.segments and time.time() - self.last_transcript_time > self.STT_SILENCE_TIMEOUT:
                    self.logger.info("STT转写空闲超时，结束本题回答")
                    return await self._end_and_save()
        except asyncio.CancelledError:
            pass

    async def _end_and_save(self):
        # 一开始检查
        if self._ended or getattr(self, "interview_ended", False):
            return
        if self._ended:
            return
        self._ended = True
        self.logger.info("本题结束，停止STT并发送DONE")
        # 停止 STT
        try:
            self.stt.nls_client.stop()
        except Exception:
            pass
        # 取消检测任务
        for t in (getattr(self, "vad_task", None), getattr(self, "stt_task", None)):
            if t:
                t.cancel()
        # 告知前端
        await self.send_json({"event": "DONE"})
        # 保存及下一题在后台
        asyncio.create_task(self._save_and_next())

    async def _save_and_next(self):
        # 一开始检查
        if getattr(self, "interview_ended", False):
            return
        if not self.segments:
            self.logger.warning("No speech segments collected — skipping save_and_next")
            await self.send_json({"event": "NO_ANSWER", "message": "未检测到有效回答，已跳过。"})
            return
        # ————— 1) 保存 Answer & AnswerAnalysis —————
        start_utc, end_utc = self.segments[0][0], self.segments[-1][0]
        text = "".join(t for _, t in self.segments)
        ans = await sync_to_async(Answer.objects.create)(
            generated_q_id=str(self.current_gq_id),
            content=text,
            start_time=start_utc,
            end_time=end_utc,
            video_path="",
            audio_path="",
        )
        await self.send_json({
            "event": "ANSWER_SAVED",
            "answer_id": str(ans.answer_id),
        })
        try:
            metrics = await analyze_answer_text(
                self.interview_id,
                self.current_question,
                text
            )
        except Exception:
            metrics = {"clarity": 0, "structure": 0, "coverage": 0, "suggestions": ["分析失败"]}
        await sync_to_async(AnswerAnalysis.objects.create)(
            interview_id=self.interview_id,
            answer_id=str(ans.answer_id),
            modality_type="audio",
            metrics=metrics,
        )
        # ===== 在生成下一题前再次检查面试是否结束 =====
        if self.interview_ended:
            self.logger.info("面试已结束，停止生成下一题")
            return
        # —— 新增：在生成下一题前先累加并检查是否达到上限 —— #
        self.sent_questions += 1
        if self.sent_questions >= self.total_questions:
            self.interview_ended = True
            # 通知前端面试结束
            await self.send_json({
                    "event": "INTERVIEW_COMPLETE",
                    "message": "题目已全部完成，面试结束，报告生成中..."
            })
            # 触发后台报表生成
            asyncio.create_task(self._run_report())
            return
        # ————— 2) 决定下一题来源 —————
        if self.use_ai_next:
            # —— AI 生成 —— #
            # 从简历和刚答内容生成 prompt
            next_q, qtype, meta = await _generate_ai_question(self.interview_id, text)
        else:
            # —— 题库抽取 —— #
            interview = await sync_to_async(Interview.objects.get)(interview_id=self.interview_id)
            asked_ids = await sync_to_async(_get_asked_question_ids)(self.interview_id)
            next_item = await sync_to_async(_pick_next_question_from_bank)(
                interview.position, asked_ids
            )
            if next_item:
                next_q, qtype = next_item.content, next_item.question_type
                meta = {
                    'question_id': next_item.question_id,
                    'difficulty': next_item.difficulty,
                    'suggested_time_min': next_item.suggested_answer_time,
                    'key_points': next_item.key_points.splitlines(),
                }
            else:
                # 题库空了，退静态
                import random
                next_q = random.choice(self.STATIC_FALLBACK_QUESTIONS)
                qtype = 'behavioral'
                meta = {'fallback': True}

        # 切换到下次模式
        self.use_ai_next = not self.use_ai_next

        # ————— 3) 存入 GeneratedQuestion 并发题 … —————
        gq_next = await sync_to_async(GeneratedQuestion.objects.create)(
            interview_id=self.interview_id,
            question_content=next_q,
            question_type=qtype,
            generation_source=json.dumps(meta, ensure_ascii=False),
            question_id=meta.get('question_id')
        )
        self.current_gq_id = gq_next.generated_q_id
        self.current_question = next_q

        await self.send_json({
            "event": "QUESTION",
            "text": f"面试 #{self.interview_id}：{next_q}",
            "qtype": qtype,
            "meta": meta,
        })
        self.logger.info(f"已发送下一题（{'题库' if self.use_ai_next else 'AI'}模式）: {next_q}")

        # ————— 4) 重置状态 —————
        self._reset_state()

    def _reset_state(self, init=False):
        """重置内部状态，init=True 时清除 STT/VAD 客户端不处理"""
        self._ended = False
        self._saved = False
        self._collect_done = False
        self.segments = []
        self.last_transcript_time = time.time()
        self.vad_silence_start = None
        if not init:
            # 取消旧任务
            for t in (getattr(self, "vad_task", None), getattr(self, "stt_task", None)):
                if t:
                    t.cancel()

    async def disconnect(self, code):
        await self.channel_layer.group_discard(f"interview_{self.interview_id}", self.channel_name)
        self.logger.info(f"WebSocket disconnect code={code}")
        # 清理 STT
        if hasattr(self, 'stt') and getattr(self.stt, 'connected', False):
            await self.stt.close()

    async def _run_report(self):
        try:
            # —— 1) 取出本次面试所有的 generated_q_id —— #
            gq_ids = await sync_to_async(_fetch_gq_ids)(self.interview_id)

            # —— 2) 等待 Answer 和 AnswerAnalysis 数量持平 —— #
            while True:
                total_answers = await sync_to_async(
                    Answer.objects.filter(generated_q_id__in=gq_ids).count
                )()
                total_analyzed = await sync_to_async(
                    AnswerAnalysis.objects.filter(interview_id=self.interview_id).count
                )()
                if total_answers == total_analyzed:
                    break

                self.logger.info(
                    f"等待分析完成：回答 {total_answers} 条，分析 {total_analyzed} 条，继续等…"
                )
                await asyncio.sleep(1.0)

            # —— 3) 全部分析到位后再生成报告 —— #
            await sync_to_async(generate_interview_report, thread_sensitive=False)(
                self.interview_id
            )

            # 获取生成的 Report
            rpt = await sync_to_async(Report.objects.get)(
                interview_id=self.interview_id
            )
            # 从 metrics JSONField 获取数据
            metrics = rpt.metrics or {}
            ordered_report = OrderedDict([
                ("overall_score", metrics.get("overall_score")),
                ("radar_data", metrics.get("radar_data", {})),
                ("strengths", metrics.get("strengths", [])),
                ("weaknesses", metrics.get("weaknesses", [])),
                ("questionDetails", metrics.get("questionDetails", [])),
                ("audio_analysis", metrics.get("audio_analysis", {})),
                ("video_analysis", metrics.get("video_analysis", {})),
                ("learning_paths", metrics.get("learning_paths", {})),
            ])
            payload = {
                "event": "REPORT_READY",
                "report": ordered_report
            }

            # 广播到 WebSocket 群组
            await get_channel_layer().group_send(
                f"interview_{self.interview_id}",
                {
                    "type": "report.ready",
                    "payload": payload
                }
            )
            self.logger.info("✅ 同步报表生成完成并推送")
        except Exception as e:
            self.logger.exception(f"❌ 报表生成失败：{e}")
            await self.send_json({"event": "ERROR", "message": "报告生成失败"})

    async def report_ready(self, event):
        # event["payload"] 里就是你想发给前端的 JSON 包
        await self.send_json(event["payload"])


def _get_asked_question_ids(interview_id):
    qs = GeneratedQuestion.objects.filter(interview_id=interview_id) \
                                  .values_list('question_id', flat=True) \
                                  .distinct()
    # 强制执行并返回 list
    return list(qs)


def _pick_next_question_from_bank(position, asked_ids):
    qs = Question.objects.filter(role=position)
    if asked_ids:
        qs = qs.exclude(pk__in=asked_ids)
    return qs.order_by('?').first()


def _fetch_gq_ids(interview_id):
    return list(
        GeneratedQuestion.objects
        .filter(interview_id=interview_id)
        .values_list('generated_q_id', flat=True)
    )